package com.chengqs.rainbow.im.route.server.command;

import cn.hutool.json.JSONUtil;
import com.chengqs.rainbow.common.im.constant.CommandType;
import com.chengqs.rainbow.common.im.constant.Constants;
import com.chengqs.rainbow.common.im.domain.MessageAckJsonRequest;
import com.chengqs.rainbow.common.protobuf.Command;
import com.chengqs.rainbow.common.protobuf.MessageAckRequest;
import io.netty.channel.ChannelHandlerContext;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

import java.util.concurrent.Executor;

/**
 * <B>主类名称：</B>MessageAckServerCommandHandler<BR>
 * <B>概要说明：</B>收到网关系统发送的客户端接受到推送消息的ack处理器<BR>
 *
 * @author: chengqs
 * @create: 2025-09-13 20:15:59
 **/
@Slf4j
@Component
@RequiredArgsConstructor
public class MessageAckServerCommandHandler implements ServerCommandHandler {
    private final KafkaTemplate<String, String> kafkaTemplate;

    @Value("#{'topic-c2c-message-ack-'.concat('${spring.profiles.active}')}")
    private String TOPIC_C2C_MESSAGE_ACK;

    @Value("#{'topic-c2g-message-ack-'.concat('${spring.profiles.active}')}")
    private String TOPIC_C2G_MESSAGE_ACK;

    private final Executor taskExecutor;

    @Override
    public void handleCommand(Command command, ChannelHandlerContext ctx) {
        taskExecutor.execute(() -> {
            try {
                MessageAckRequest body = MessageAckRequest.parseFrom(command.getBody());

                MessageAckJsonRequest request = new MessageAckJsonRequest();
                request.setChatType(body.getChatType()); // 聊天类型（单聊、群聊）
                request.setChatId(body.getChatId()); // 聊天id
                request.setMemberId(body.getMemberId()); // 谁返回的ack
                request.setClientId(command.getClient()); // 客户端（手机、平板、电脑等）
                request.setMessageId(body.getMessageId()); // 消息唯一标识（这是候一般指最新的消息）
                if (Constants.CHAT_TYPE_C2G.equals(request.getChatType())) {
                    kafkaTemplate.send(TOPIC_C2G_MESSAGE_ACK, JSONUtil.toJsonStr(request));
                } else {
                    kafkaTemplate.send(TOPIC_C2C_MESSAGE_ACK, JSONUtil.toJsonStr(request));
                }
            } catch (Exception e) {
                log.error(e.getMessage(), e);
            }
        });
    }

    @Override
    public int getType() {
        return CommandType.COMMAND_MESSAGE_ACK;
    }
}
